#include <zmq.h>
#include <stdio.h>
#include <string.h>

typedef struct
{
    zmq_msg_t srcaddr;
    zmq_msg_t dstaddr;
    zmq_msg_t data;
}router_msg_t;

//#define COMM_URL "tcp://127.0.0.1:5555"
#define COMM_URL "inproc://router_test"

static void dump_data2(const char* module, const char* header, const void* data, int size)
{
    unsigned char* ptr = (unsigned char*)data;

    printf("%s:%s: %.*s\r\n", module, header, size, ptr);
}


int router_recv_msg(void* fd, router_msg_t* rmsg)
{
    int ret;

    zmq_msg_init(&rmsg->dstaddr);
    zmq_msg_init(&rmsg->srcaddr);
    zmq_msg_init(&rmsg->data);

    //这里的消息是router自己添加的发送方的routing ID
    ret = zmq_msg_recv(&rmsg->srcaddr, fd, 0);
    //dump_address(zmq_msg_data(&address), ret);
    if (ret < 0) {
        return ret;
    }

    //这里的消息是发送方自己添加的目的方的routing ID
    if (zmq_msg_more(&rmsg->srcaddr))
    {
        ret = zmq_msg_recv(&rmsg->dstaddr, fd, 0);
        if (ret < 0) {
            zmq_msg_close(&rmsg->srcaddr);
            return ret;
        }
    }

    //这里的消息是最终要发送的消息内容
    if (zmq_msg_more(&rmsg->dstaddr))
    {
        ret = zmq_msg_recv(&rmsg->data, fd, 0);
        if (ret < 0) {
            zmq_msg_close(&rmsg->srcaddr);
            zmq_msg_close(&rmsg->dstaddr);
            return ret;
        }
        
        return 0;
    }
    
    return -1;
}

int router_handle_msg(void* fd, router_msg_t* rmsg)
{
    int ret;

    ret =  zmq_msg_send(&rmsg->dstaddr, fd, ZMQ_SNDMORE);
    ret |= zmq_msg_send(&rmsg->srcaddr, fd, ZMQ_SNDMORE);
    ret |= zmq_msg_send(&rmsg->data,    fd, 0);

    return ret;
}

void client_recv_reply(const char *owner, void* fd)
{
    zmq_msg_t address;
    zmq_msg_t data;
    int ret;

    zmq_msg_init(&address);
    zmq_msg_init(&data);

    ret = zmq_msg_recv(&address, fd, 0);

    if (ret < 0) {
        return;
    }

    dump_data2(owner, "srcaddr: ", zmq_msg_data(&address), ret);

    /*从第二帧开始才是对端发来的消息*/
    if (zmq_msg_more(&address))
    {
        ret = zmq_msg_recv(&data, fd, 0);
    }

    //printf("ret = %d...\r\n", ret);
    if (ret > 0)
    {
        dump_data2(owner, "rcvdata: ", zmq_msg_data(&data), ret);
    }

    zmq_msg_close(&address);
    zmq_msg_close(&data);
}

void req_main(void* arg)
{
    void* context = arg;
    void* dealer = zmq_socket(context, ZMQ_DEALER);

    zmq_setsockopt(dealer, ZMQ_ROUTING_ID, "ATEStudio", 9);
    int recvTime = 20;
    zmq_setsockopt(dealer, ZMQ_RCVTIMEO, &recvTime, sizeof(recvTime));
    zmq_connect(dealer, COMM_URL);
    zmq_sleep(1);

    int  ret;
    char buf[256] = "hello";
    int  total = 10;

    while(total--)
    {
        zmq_send_const(dealer, "SDKMaster", 9, ZMQ_SNDMORE);
        zmq_send(dealer, "data", 4, 0);
        client_recv_reply("ATEStudio", dealer);
        zmq_sleep(1);
    }

    zmq_sleep(1);
    zmq_close(dealer);
}

void cmd_main(void* arg)
{
    void* context = arg;
    void* dealer = zmq_socket(context, ZMQ_DEALER);

    zmq_setsockopt(dealer, ZMQ_ROUTING_ID, "cli", 3);
    int recvTime = 20;
    zmq_setsockopt(dealer, ZMQ_RCVTIMEO, &recvTime, sizeof(recvTime));
    zmq_connect(dealer, COMM_URL);
    zmq_sleep(1);

    int  ret;
    char buf[256] = "hello";
    int  total = 10;

    while (total--)
    {
        zmq_send_const(dealer, "SDKSlave", 8, ZMQ_SNDMORE);
        zmq_send(dealer, "data", 4, 0);
        client_recv_reply("cli", dealer);
        zmq_sleep(1);
    }

    zmq_sleep(1);
    zmq_close(dealer);
}

void server_recv_request(const char *owner, void* fd)
{
    zmq_msg_t address;
    zmq_msg_t data;
    int ret;

    zmq_msg_init(&address);
    zmq_msg_init(&data);

    ret = zmq_msg_recv(&address, fd, 0);

    if (ret < 0) {
        return;
    }
    dump_data2(owner, "srcaddr: ", zmq_msg_data(&address), ret);

    /*从第二帧开始才是对端发来的消息*/
    if (zmq_msg_more(&address))
    {
        ret = zmq_msg_recv(&data, fd, 0);
    }

    //printf("ret = %d...\r\n", ret);
    if (ret > 0)
    {
        dump_data2(owner, "rcvdata: ", zmq_msg_data(&data), ret);
    }

    zmq_msg_send(&address, fd, ZMQ_SNDMORE);
    zmq_send(fd, "reply msg...", 12, 0);

    zmq_msg_close(&address);
    zmq_msg_close(&data);
}


void rep_master_main(void* arg)
{
    void* context = arg;
    void* dealer = zmq_socket(context, ZMQ_DEALER);

    zmq_setsockopt(dealer, ZMQ_ROUTING_ID, "SDKMaster", 9);
    int recvTime = -1;
    zmq_setsockopt(dealer, ZMQ_RCVTIMEO, &recvTime, sizeof(recvTime));
    zmq_connect(dealer, COMM_URL);
    zmq_sleep(1);

    while (1) {
        server_recv_request("SDKMaster", dealer);
    }

    zmq_close(dealer);
    zmq_ctx_destroy(context);
}

void rep_slave_main(void* arg)
{
    void* context = arg;
    void* dealer = zmq_socket(context, ZMQ_DEALER);

    zmq_setsockopt(dealer, ZMQ_ROUTING_ID, "SDKSlave", 8);
    int recvTime = -1;
    zmq_setsockopt(dealer, ZMQ_RCVTIMEO, &recvTime, sizeof(recvTime));
    zmq_connect(dealer, COMM_URL);
    zmq_sleep(1);

    while (1) {
        server_recv_request("SDKSlave", dealer);
    }

    zmq_close(dealer);
    zmq_ctx_destroy(context);
}


int main(void)
{
    void* context = zmq_ctx_new();
    void* router = zmq_socket(context, ZMQ_ROUTER);
    zmq_bind(router, COMM_URL);

    zmq_threadstart(req_main, context);
    zmq_threadstart(cmd_main, context);
    zmq_threadstart(rep_master_main, context);
    zmq_threadstart(rep_slave_main,  context);

    router_msg_t rmsg;
    int ret;

    while (1)
    {
        ret = router_recv_msg(router, &rmsg);
        if (ret >= 0) {
            router_handle_msg(router, &rmsg);
        }
    }
    zmq_close(router);
    zmq_ctx_destroy(context);
    return 0;
}